/*-
 * Copyright 2012 Guram Dukashvili
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
//---------------------------------------------------------------------------
#include "sys.h"
#include "str.h"
#include "idx.h"
//---------------------------------------------------------------------------
static const struct signature journal_block_signature = {
  {
    0x11, 0xD3, 0x40, 0x16, 0x15, 0x93, 0x52, 0xF8,
    0xC0, 0x2E, 0x2A, 0xE8, 0x43, 0xB7, 0x50, 0xC4,
    0x1B, 0xB5, 0x1D, 0x37, 0xBC, 0x47, 0x49, 0x41,
    0xE4, 0x55, 0x6D, 0x29, 0x7B, 0x75, 0x9D, 0xFD
  }
};
//---------------------------------------------------------------------------
void index_write_journal(struct index * idx)
{
  if( idx->changed_links.count > 0 ){
    int file_handle;
    portable_ssize_t w;
    intptr_t p;
    uintptr_t insert_position, lnk, lnk2, count;
    string filter;
    array changed_links, dir_list;
    struct cdc256_data checksum_data;
    struct index_link_change * plnk;
    struct index_link * link;
    struct index_block * blk;
    struct index_journal_block_state * state;
    uintptr_t jdata_size, state_size, jidx_size, jhdr_size;
    struct index_journal_block * jhdr;

    rwlock_acquire_write_lock(&idx->changed_links_lock);

    array_init_copy(&changed_links,&idx->changed_links,0);

    plnk = (void *) changed_links.a;

    for( lnk = count = 0; lnk < changed_links.count; lnk++ ){
      if( plnk[lnk].what != ILC_DEL ) count++;
    }

    jdata_size = count * idx->cache->block_size;
    state_size = sizeof(struct index_journal_block_state) - sizeof(byte_size_int) + idx->block_index_size;
    jidx_size = state_size * changed_links.count;
    jhdr_size = sizeof(struct index_journal_block) + jidx_size - state_size;

    jhdr_size += -jhdr_size & (idx->cache->block_size - 1);
    jhdr = salloc(jhdr_size + jdata_size,NULL);

    memcpy(&jhdr->signature,&journal_block_signature,sizeof(jhdr->signature));

    // pass 1 - just copy
    for( lnk2 = lnk = 0; lnk < changed_links.count; lnk++ ){
      link = plnk[lnk].link;

      state = (void *) ((uint8_t *) jhdr->blocks + lnk * state_size);
      state->what = plnk[lnk].what;
      memcpy(&state->block_index,&link->block_index,idx->block_index_size);

      mutex_acquire(&idx->cache->links_mutex);
#if COMPILE_RUNTIME_CHECKS
      if( link->list != &idx->cache->links_linked ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
      link->linked++;
      mutex_release(&idx->cache->links_mutex);

      if( plnk[lnk].what != ILC_DEL ){
        blk = (void *) ((uint8_t *) jhdr + jhdr_size + lnk2 * idx->cache->block_size);
        memcpy(blk,&link->blk,idx->cache->block_size);
        blk->count = htobe32(blk->count);

        lnk2++;
      }
    }

    array_resize(&idx->changed_links,0);

    rwlock_release_write_lock(&idx->changed_links_lock);

    // pass 2 - compute blocks checksum
    for( lnk2 = lnk = 0; lnk < changed_links.count; lnk++ ){
      link = plnk[lnk].link;

      if( plnk[lnk].what != ILC_DEL ){
        blk = (void *) ((uint8_t *) jhdr + jhdr_size + lnk2 * idx->cache->block_size);

        cdc256_init(&checksum_data);
        cdc256_update(&checksum_data,&blk->checksum + 1,idx->cache->block_size - sizeof(blk->checksum));
        cdc256_final(&checksum_data,&blk->checksum);

        lnk2++;
      }
    }

    jhdr->block_count = htobe32(changed_links.count);
#if COMPILE_RUNTIME_CHECKS
    {
      uint8_t * p = (uint8_t *) jhdr->blocks + changed_links.count * state_size;
      memset(p,0,(uint8_t *) jhdr + jhdr_size - p);
    }
#endif

    cdc256_init(&checksum_data);
    cdc256_update(&checksum_data,
      &jhdr->checksum + 1,
      jhdr_size + jdata_size - sizeof(jhdr->checksum));
    cdc256_final(&checksum_data,&jhdr->checksum);

    string_init(&filter);

    index_get_file_filter(idx,INDEX_JRN_FILE_HANDLE,&filter,idx->mvcc_id);

    dir_list_init(&dir_list);
    dir_list_get(&dir_list,filter.s,false,true);

    file_handle = index_get_file_handle(idx,INDEX_JRN_FILE_HANDLE,dir_list.count,true);

    w = portable_pwrite(file_handle,jhdr,jhdr_size + jdata_size,0);
    if( w < 0 ){
      perror(NULL);
#if COMPILE_RUNTIME_CHECKS
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
      abort();
    }

    portable_fsync(file_handle);
    portable_close(file_handle);

    for( lnk2 = lnk = 0; lnk < changed_links.count; lnk++ ){
      struct index_block_mapper_entry bmap;
      enum INDEX_LINK_CHANGE what = plnk[lnk].what;
      enum RWLOCK_TYPE lock_type;

      link = plnk[lnk].link;

      memcpy(&bmap.block_index,&link->block_index,idx->block_index_size);
      bsint_set_htobe64(&bmap.jrn_block_index,idx->block_index_size,jhdr_size / idx->cache->block_size + lnk2);
      bmap.jrn_file_index = dir_list.count;

      rwlock_acquire_read_lock(&idx->blocks_map_lock);
      lock_type = RWL_READ;

      p = array_lookup(&idx->blocks_map,&bmap,&insert_position);

      if( p < 0 ){
        if( what != ILC_DEL ){
          rwlock_upgrade(&idx->blocks_map_lock);
          lock_type = RWL_WRITE;
          array_insert_copy(&idx->blocks_map,insert_position,&bmap);
        }
      }
      else if( what == ILC_DEL ){
        rwlock_upgrade(&idx->blocks_map_lock);
        lock_type = RWL_WRITE;
        array_remove(&idx->blocks_map,p);
      }
      else {
        struct index_block_mapper_entry * m = array_get(&idx->blocks_map,p);
        rwlock_upgrade(&idx->blocks_map_lock);
        lock_type = RWL_WRITE;
        memcpy(&m->jrn_block_index,&bmap.jrn_block_index,idx->block_index_size);
        m->jrn_file_index = bmap.jrn_file_index;
      }

      rwlock_release(&idx->blocks_map_lock,lock_type);

      rwlock_acquire_read_lock(&idx->changed_links_lock);
      mutex_acquire(&idx->cache->links_mutex);
#if COMPILE_RUNTIME_CHECKS
      if( link->list != &idx->cache->links_linked ){
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
        abort();
      }
#endif
      if( --link->linked == 0 ){
        if( array_lookup(&idx->changed_links,plnk + lnk,&insert_position) < 0 ){
          list_remove(&idx->cache->links_linked,link);
          list_insert_head(&idx->cache->links_lru,link);
          link->list = &idx->cache->links_lru;
        }
      }
      mutex_release(&idx->cache->links_mutex);
      rwlock_release_read_lock(&idx->changed_links_lock);

      if( what != ILC_DEL ) lnk2++;
    }

    sfree(jhdr);

    array_destroy(&dir_list);
    string_destroy(&filter);

    array_destroy(&changed_links);
  }
}
//---------------------------------------------------------------------------
void index_write_journal_entry(struct thread * thr,struct index * idx)
{
  while( !thr->terminate ){
    semaphore_timed_wait(&idx->write_journal_wait,30000000);
    index_write_journal(idx);
  }
}
//---------------------------------------------------------------------------
void index_flush_journal(struct index * idx)
{
  int jrn_file_index;
  string filter;
  array dir_list;
  array * bitmap_mod = alloca(sizeof(array) +
    sizeof(struct index_journal_block_state *) * 64);
#if COMPILE_RUNTIME_CHECKS
  void * zero = alloca(idx->cache->block_size);
  memset(zero,0,idx->cache->block_size);
#endif

  array_init(bitmap_mod);
  array_set_element_size(
    bitmap_mod,
    sizeof(struct index_journal_block_state *),
    sizeof(struct index_journal_block_state *) * 64);

  string_init(&filter);

  index_get_file_filter(idx,INDEX_JRN_FILE_HANDLE,&filter,idx->mvcc_id);

  dir_list_init(&dir_list);
  dir_list_get(&dir_list,filter.s,false,true);

  for( jrn_file_index = 0; (uintptr_t) jrn_file_index < dir_list.count; jrn_file_index++ ){
    struct index_block * blk;
    struct index_journal_block_state * state;
    int jrn_file_handle, file_handle;
    intptr_t ilnk, p;
    uintptr_t insert_position, count, state_size, jidx_size, jhdr_size, lnk, lnk2;
    portable_ssize_t r, w;
    portable_stat_t st;
    portable_off_t off;
    struct cdc256_data checksum_data;
    struct cdc256 checksum;
    struct index_journal_block * jhdr;
    void * bitmap_block_buf = NULL;
    uint64_t extent_size = idx->cache->block_size * 8;

    string * name = array_get(&dir_list,jrn_file_index);

    if( portable_stat(name->s,&st) != 0
      || st.st_size % idx->cache->block_size != 0 ) break;

    jrn_file_handle = index_get_file_handle(idx,INDEX_JRN_FILE_HANDLE,jrn_file_index,false);

    jhdr = salloc(st.st_size,NULL);

    r = portable_pread(jrn_file_handle,jhdr,st.st_size,0);
    if( r < 0 ){
      perror(NULL);
#if COMPILE_RUNTIME_CHECKS
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
      abort();
    }

    portable_close(jrn_file_handle);

    if( memcmp(&jhdr->signature,&journal_block_signature,sizeof(jhdr->signature)) != 0 ) break;

    cdc256_init(&checksum_data);
    cdc256_update(&checksum_data,&jhdr->checksum + 1,st.st_size - sizeof(jhdr->checksum));
    cdc256_final(&checksum_data,&checksum);

    if( memcmp(&jhdr->checksum,&checksum,sizeof(jhdr->checksum)) != 0 ) break;

    count = be32toh(jhdr->block_count);
    state_size = sizeof(struct index_journal_block_state) - sizeof(byte_size_int) + idx->block_index_size;
    jidx_size = state_size * count;
    jhdr_size = sizeof(struct index_journal_block) + jidx_size - state_size;

    jhdr_size += -jhdr_size & (idx->cache->block_size - 1);

    file_handle = index_get_file_handle(idx,INDEX_IDX_FILE_HANDLE,0,true);

    for( lnk2 = lnk = 0; lnk < count; lnk++ ){
      state = (void *) ((uint8_t *) jhdr->blocks + lnk * state_size);

      if( state->what == ILC_NEW || state->what == ILC_DEL )
        array_append_copy(bitmap_mod,&state);

#if !COMPILE_RUNTIME_CHECKS
      if( state->what != ILC_DEL )
#endif
      {
        blk = (void *) ((uint8_t *) jhdr + jhdr_size + lnk2 * idx->cache->block_size);
#if COMPILE_RUNTIME_CHECKS
        if( state->what == ILC_DEL ) blk = zero;
#endif
        off = (portable_off_t) bsint_get_be64toh(&state->block_index,idx->block_index_size)
          * idx->cache->block_size;
        w = portable_pwrite(file_handle,blk,idx->cache->block_size,off);
        if( w != (portable_ssize_t) idx->cache->block_size ){
          perror(NULL);
#if COMPILE_RUNTIME_CHECKS
          fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
          abort();
        }
#if COMPILE_RUNTIME_CHECKS
        if( state->what != ILC_DEL )
#endif
        lnk2++;
      }
    }

    lnk = 0;
    while( lnk < bitmap_mod->count ){

      struct index_journal_block_state * pbitmap_mod =
        *(struct index_journal_block_state **) array_get(bitmap_mod,lnk);

      uint64_t block_index = bsint_get_be64toh(&pbitmap_mod->block_index,idx->block_index_size);
      uint64_t block_extent = block_index / extent_size;
      uint64_t extent_block_index = block_extent == 0 ? 2 : block_extent * extent_size;

      if( bitmap_block_buf == NULL ) bitmap_block_buf = salloc(idx->cache->block_size,NULL);

      off = extent_block_index * idx->cache->block_size;
      r = portable_pread(file_handle,bitmap_block_buf,idx->cache->block_size,off);
      if( r != (portable_ssize_t) idx->cache->block_size ){
        perror(NULL);
#if COMPILE_RUNTIME_CHECKS
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
        abort();
      }

      set_bit_range(bitmap_block_buf,0,block_extent == 0 ? 3 : 1);

      for( lnk2 = lnk; lnk2 < bitmap_mod->count; lnk2++ ){
        struct index_journal_block_state * pbitmap_mod2 =
          *(struct index_journal_block_state **) array_get(bitmap_mod,lnk2);

        uint64_t block_index2 = bsint_get_be64toh(&pbitmap_mod2->block_index,idx->block_index_size);
        uint64_t block_extent2 = block_index2 / extent_size;
        if( block_extent != block_extent2 ) break;

        if( pbitmap_mod2->what == ILC_NEW )
          set_bit(bitmap_block_buf,block_index2 & (extent_size - 1));
        else if( pbitmap_mod2->what == ILC_DEL )
          reset_bit(bitmap_block_buf,block_index2 & (extent_size - 1));
      }

      lnk = lnk2;

      off = extent_block_index * idx->cache->block_size;
      w = portable_pwrite(file_handle,bitmap_block_buf,idx->cache->block_size,off);

      if( w != (portable_ssize_t) idx->cache->block_size ){
        perror(NULL);
#if COMPILE_RUNTIME_CHECKS
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
        abort();
      }
    }

    sfree(bitmap_block_buf);
    array_resize(bitmap_mod,0);

    portable_fsync(file_handle);

    index_put_file_handle(idx,INDEX_IDX_FILE_HANDLE,0,file_handle);

    for( ilnk = count - 1; ilnk >= 0; ilnk-- ){
      struct index_journal_block_state * state =
        (void *) ((uint8_t *) jhdr->blocks + ilnk * state_size);

      struct index_block_mapper_entry bmap;

      memcpy(&bmap.block_index,&state->block_index,idx->block_index_size);

      rwlock_acquire_write_lock(&idx->blocks_map_lock);
      
      p = array_lookup(&idx->blocks_map,&bmap,&insert_position);

      if( p >= 0 ){
        struct index_block_mapper_entry * m = array_get(&idx->blocks_map,p);
        if( m->jrn_file_index == jrn_file_index )
          array_remove(&idx->blocks_map,p);
      }

      rwlock_release_write_lock(&idx->blocks_map_lock);
    }

    sfree(jhdr);

    index_close_file_handles(idx,INDEX_JRN_FILE_HANDLE,jrn_file_index);
    if( portable_unlink(name->s) != 0 ){
      perror(NULL);
#if COMPILE_RUNTIME_CHECKS
      fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
      abort();
    }
  }

  while( (uintptr_t) jrn_file_index < dir_list.count ){ // drop remaning journal records if dirty shutdown detected

    string * name = array_get(&dir_list,jrn_file_index);

    index_close_file_handles(idx,INDEX_JRN_FILE_HANDLE,jrn_file_index);
    if( portable_unlink(name->s) != 0 ){
        perror(NULL);
#if COMPILE_RUNTIME_CHECKS
        fprintf(stderr,"undefined behavior in %s, line %d\n",__FILE__,__LINE__);
#endif
        abort();
      }

    jrn_file_index++;
  }

  array_destroy(&dir_list);
  string_destroy(&filter);
  array_destroy(bitmap_mod);
}
//---------------------------------------------------------------------------
